package org.yeauty.pojo;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.yeauty.standard.ServerEndpointConfig;

import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Yeauty
 * @version 1.0
 */
@Slf4j
public class PojoEndpointServer {
    private static ExecutorService executorService;

    private static ThreadLocal<Session> sessionThreadLocal = new ThreadLocal<>();

//    private static final AttributeKey POJO_KEY = AttributeKey.valueOf("WEBSOCKET_IMPLEMENT");

    private static final AttributeKey<Session> SESSION_KEY = AttributeKey.valueOf("WEBSOCKET_SESSION");

    private static final AttributeKey<String> PATH_KEY = AttributeKey.valueOf("WEBSOCKET_PATH");


    private final ApplicationContext applicationContext;

    private final ServerEndpointConfig config;


    private boolean isOpened = false;

    public PojoEndpointServer(ApplicationContext applicationContext, ServerEndpointConfig config) {
        this.applicationContext = applicationContext;
        this.config = config;

        executorService = new ThreadPoolExecutor(config.getWORK_POOL_SIZE(), config.getWORK_POOL_SIZE(),
                0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(config.getWORK_POOL_QUEUE_SIZE()), // 使用有界队列，避免OOM
                (Runnable r, ThreadPoolExecutor executor) -> {
                    Session session = sessionThreadLocal.get();
                    applicationContext.getBean(WorkPoolRejectedExecutionSessionHandler.class)
                            .rejectedExecution(session, r, executor);
                });
    }

    public void doOnOpen(ChannelHandlerContext ctx, FullHttpRequest req, String path, String originalParam) {
        Channel channel = ctx.channel();
        Attribute<String> attrPath = channel.attr(PATH_KEY);
        attrPath.set(path);
        Attribute<Session> attrSession = channel.attr(SESSION_KEY);
        Session session = null;
        try {
            session = new Session(channel);
            attrSession.set(session);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        isOpened = true;
        HttpHeaders headers = req.headers();
        try {
            WebsocketHandler handler = this.applicationContext.getBean(path, WebsocketHandler.class);
            ParameterMap parameterMap = new ParameterMap(originalParam);
            handler.onOpen(session, headers, parameterMap);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    public void doOnClose(ChannelHandlerContext ctx) {
        if (!isOpened) {
            return;
        }
        Attribute<String> attrPath = ctx.channel().attr(PATH_KEY);
        Session session = ctx.channel().attr(SESSION_KEY).get();
        if (session == null) {
            log.error("session is null");
            return;
        }
        try {
            WebsocketHandler handler = this.applicationContext.getBean(attrPath.get(), WebsocketHandler.class);
            handler.onClose(session);
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
        }
//        }
    }


    public void doOnError(ChannelHandlerContext ctx, Throwable throwable) {
        if (!isOpened) {
            return;
        }
        Attribute<String> attrPath = ctx.channel().attr(PATH_KEY);
        Session session = ctx.channel().attr(SESSION_KEY).get();
        if (session == null) {
            log.error("session is null");
            return;
        }
        try {
            WebsocketHandler handler = this.applicationContext.getBean(attrPath.get(), WebsocketHandler.class);
            handler.onError(session, throwable);
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
        }
    }

    public void doOnMessage(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (!isOpened) {
            return;
        }
        Attribute<String> attrPath = ctx.channel().attr(PATH_KEY);
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        Session session = ctx.channel().attr(SESSION_KEY).get();
        try {
            WebsocketHandler handler = this.applicationContext.getBean(attrPath.get(), WebsocketHandler.class);
            handler.onMessage(session, textFrame.text());
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
        }
    }

    public void doOnBinary(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (!isOpened) {
            return;
        }
        Attribute<String> attrPath = ctx.channel().attr(PATH_KEY);
        BinaryWebSocketFrame binaryWebSocketFrame = (BinaryWebSocketFrame) frame;
        ByteBuf content = binaryWebSocketFrame.content();
        byte[] bytes = new byte[content.readableBytes()];
        content.readBytes(bytes);
        Session session = ctx.channel().attr(SESSION_KEY).get();

        sessionThreadLocal.set(session);

        executorService.execute(() -> {
            try {
                WebsocketHandler handler = this.applicationContext.getBean(attrPath.get(), WebsocketHandler.class);
                handler.onBinary(session ,bytes);
            } catch (Throwable t) {
                log.error(t.getMessage(), t);
            }
        });
//        }
    }

    public void doOnEvent(ChannelHandlerContext ctx, Object evt) {
        if (!isOpened) {
            return;
        }
        Attribute<String> attrPath = ctx.channel().attr(PATH_KEY);
            Session session = ctx.channel().attr(SESSION_KEY).get();
            try {
                WebsocketHandler handler = this.applicationContext.getBean(attrPath.get(), WebsocketHandler.class);
                handler.onEvent(session ,evt);
            } catch (Throwable t) {
                log.error(t.getMessage(), t);
            }
//        }
    }

    public String getHost() {
        return config.getHost();
    }

    public Set<String> getPathSet() {
        return config.getPathSet();
    }

}
